#define _GNU_SOURCE
#include "app_device.h"
#include "fcntl.h"
#include "app_pool.h"
#include "app_mqtt.h"
#include "log/log.h"
#include "stdlib.h"
#include "unistd.h"
#include "app_message.h"
#include "app_common.h"

#define MAX_BUFFER_SIZE 1024
#define POOL_SIZE 5
#define INTERVAL_TIME 200

static Device_t *device;

static int send_task_func(void *arg)
{
    char data_buffer[128];

    int len = app_buffer_read(device->up_buffer, data_buffer, sizeof(data_buffer));

    if(len  <= 0 )
    {
        log_error("read data failed");
       return -1;
    }   

    char *json = app_message_charsToJson(data_buffer);

    if(json == NULL)
    {
        log_error("json is null");
        free(json);
        return -1;
    }

    int res = app_mqtt_send(json);

    if(res != 0)
    {
        log_error("mqtt send failed");
        free(json);
        return -1;
    }

    free(json);

    return 0;
}

static void *read_thread_func(void *arg)
{
    while(device->is_running)
    {
        char data_buffer[128];

        int len = read(device->fd, data_buffer, sizeof(data_buffer));

        if(len > 0 && device->post_read != NULL)
        {
            len = device->post_read(data_buffer, len);

            if(len > 0)
            {
                app_buffer_write(device->up_buffer, data_buffer, len);

                app_pool_registerMqTask(send_task_func, NULL);
            }
        }

        // if(len > 0)
        //     {
        //         app_buffer_write(device->up_buffer, data_buffer, len);

        //         app_pool_registerMqTask(send_task_func, NULL);
        //     }
    }
}

static int write_task_func(void *arg)
{
    char data_buffer[128];

    int len = app_buffer_read(device->down_buffer, data_buffer, sizeof(data_buffer));

    if(len > 0  && device->pre_write != NULL)
    {
        len = device->pre_write(data_buffer, len);

        long interval = app_common_getCurrentTime() - device->last_write_time;

        if(len > 0 && (interval > INTERVAL_TIME))
        {
            write(device->fd, data_buffer, len);

            device->last_write_time = app_common_getCurrentTime();
        }
    }

    // long interval = app_common_getCurrentTime() - device->last_write_time;

    //     if(len > 0 && (interval > INTERVAL_TIME))
    //     {
    //         write(device->fd, data_buffer, len);

    //         device->last_write_time = app_common_getCurrentTime();
    //     }

}
static int receive_msg_callback(char *json)
{
    char data_buffer[128];

    int len = app_message_jsonToChars(json, data_buffer , sizeof(data_buffer));

    if(len <= 0)
    {
        return -1;
    }

    app_buffer_write(device->down_buffer, data_buffer, len);

    app_pool_registerMqTask(write_task_func, NULL);

    return 0;
}

Device_t *app_device_init(char *filename)
{
    if(device != NULL)
    {
        return device;
    }
    device = (Device_t *)malloc(sizeof(Device_t));

    device->filename = filename;
    device->fd=open(filename, O_RDWR);

    device->up_buffer = app_buffer_init(MAX_BUFFER_SIZE);
    device->down_buffer = app_buffer_init(MAX_BUFFER_SIZE);

    device->post_read = NULL;
    device->pre_write = NULL;

    device->last_write_time=0;

    app_pool_init(POOL_SIZE);

    app_mqtt_init();

    return device;
}

int app_device_start(void)
{
    if(device->is_running)
    {
        log_debug("device is running");
        return 0;
    }

    device->is_running = 1;

    int res = pthread_create(&device->read_thread, NULL,read_thread_func,NULL);

    if(res != 0)
    {
        log_error("create read thread failed");
        return -1;
    }

    app_mqtt_registerRevCallback(receive_msg_callback);

    return 0;
}

void app_device_destroy(void)
{
    close(device->fd);

    app_buffer_destroy(device->up_buffer);
    app_buffer_destroy(device->down_buffer);

    pthread_cancel(device->read_thread);

    device->is_running = 0;

    device->post_read = NULL;
    device->pre_write = NULL;

    app_pool_destroy();

    app_mqtt_close();

    free(device);
}